{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyalink.alink import *\n",
    "useLocalEnv(1)\n",
    "\n",
    "from utils import *\n",
    "import os\n",
    "import pandas as pd\n",
    "\n",
    "pd.set_option('display.max_colwidth', 1000)\n",
    "pd.set_option('display.max_rows', 200)\n",
    "\n",
    "DATA_DIR = ROOT_DIR + \"iris\" + os.sep\n",
    "\n",
    "ORIGIN_FILE = \"iris.data\";\n",
    "\n",
    "TRAIN_FILE = \"train.ak\";\n",
    "TEST_FILE = \"test.ak\";\n",
    "\n",
    "SCHEMA_STRING = \"sepal_length double, sepal_width double, \"\\\n",
    "                + \"petal_length double, petal_width double, category string\"\n",
    "\n",
    "FEATURE_COL_NAMES = [\"sepal_length\", \"sepal_width\", \"petal_length\", \"petal_width\"]\n",
    "\n",
    "LABEL_COL_NAME = \"category\";\n",
    "VECTOR_COL_NAME = \"vec\";\n",
    "PREDICTION_COL_NAME = \"pred\";\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_1_1\n",
    "\n",
    "source = CsvSourceBatchOp()\\\n",
    "    .setFilePath(DATA_DIR + ORIGIN_FILE)\\\n",
    "    .setSchemaStr(SCHEMA_STRING);\n",
    "\n",
    "source\\\n",
    "    .link(\n",
    "        FirstNBatchOp().setSize(5)\n",
    "    )\\\n",
    "    .print();\n",
    "\n",
    "source.firstN(5).print();\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_1_2\n",
    "\n",
    "source = CsvSourceBatchOp()\\\n",
    "    .setFilePath(DATA_DIR + ORIGIN_FILE)\\\n",
    "    .setSchemaStr(SCHEMA_STRING);\n",
    "\n",
    "source\\\n",
    "    .sampleWithSize(50)\\\n",
    "    .lazyPrintStatistics(\"< after sample with size 50 >\")\\\n",
    "    .sample(0.1)\\\n",
    "    .print();\n",
    "\n",
    "source\\\n",
    "    .lazyPrintStatistics(\"< origin data >\")\\\n",
    "    .sampleWithSize(150, True)\\\n",
    "    .lazyPrintStatistics(\"< after sample with size 150 >\")\\\n",
    "    .sample(0.03, True)\\\n",
    "    .print();\n",
    "\n",
    "source_stream = CsvSourceStreamOp()\\\n",
    "    .setFilePath(DATA_DIR + ORIGIN_FILE)\\\n",
    "    .setSchemaStr(SCHEMA_STRING);\n",
    "\n",
    "source_stream.sample(0.1).print();\n",
    "\n",
    "StreamOperator.execute();\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# c_1_3\n",
    "\n",
    "source = CsvSourceBatchOp()\\\n",
    "    .setFilePath(DATA_DIR + ORIGIN_FILE)\\\n",
    "    .setSchemaStr(SCHEMA_STRING);\n",
    "\n",
    "source\\\n",
    "    .select(\"*, CASE category WHEN 'Iris-versicolor' THEN 1 \"\n",
    "            + \"WHEN 'Iris-setosa' THEN 2 ELSE 4 END AS weight\")\\\n",
    "    .link(\n",
    "        WeightSampleBatchOp()\\\n",
    "            .setRatio(0.4)\\\n",
    "            .setWeightCol(\"weight\")\n",
    "    )\\\n",
    "    .groupBy(\"category\", \"category, COUNT(*) AS cnt\")\\\n",
    "    .print();\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_1_4\n",
    "\n",
    "source = CsvSourceBatchOp()\\\n",
    "    .setFilePath(DATA_DIR + ORIGIN_FILE)\\\n",
    "    .setSchemaStr(SCHEMA_STRING);\n",
    "\n",
    "source\\\n",
    "    .link(\n",
    "        StratifiedSampleBatchOp()\\\n",
    "            .setStrataCol(\"category\")\\\n",
    "            .setStrataRatios(\"Iris-versicolor:0.2,Iris-setosa:0.4,Iris-virginica:0.8\")\n",
    "    )\\\n",
    "    .groupBy(\"category\", \"category, COUNT(*) AS cnt\")\\\n",
    "    .print();\n",
    "\n",
    "source_stream = CsvSourceStreamOp()\\\n",
    "    .setFilePath(DATA_DIR + ORIGIN_FILE)\\\n",
    "    .setSchemaStr(SCHEMA_STRING);\n",
    "\n",
    "source_stream\\\n",
    "    .link(\n",
    "        StratifiedSampleStreamOp()\\\n",
    "            .setStrataCol(\"category\")\\\n",
    "            .setStrataRatios(\"Iris-versicolor:0.2,Iris-setosa:0.4,Iris-virginica:0.8\")\n",
    "    )\\\n",
    "    .print();\n",
    "\n",
    "StreamOperator.execute();\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_2\n",
    "\n",
    "source = CsvSourceBatchOp()\\\n",
    "    .setFilePath(DATA_DIR + ORIGIN_FILE)\\\n",
    "    .setSchemaStr(SCHEMA_STRING);\n",
    "\n",
    "print(\"schema of source:\");\n",
    "print(source.getColNames());\n",
    "\n",
    "spliter = SplitBatchOp().setFraction(0.9);\n",
    "\n",
    "source.link(spliter);\n",
    "\n",
    "print(\"schema of spliter's main output:\");\n",
    "print(spliter.getColNames());\n",
    "\n",
    "print(\"count of spliter's side outputs:\");\n",
    "print(spliter.getSideOutputCount());\n",
    "\n",
    "print(\"schema of spliter's side output :\");\n",
    "print(spliter.getSideOutput(0).getColNames());\n",
    "\n",
    "spliter\\\n",
    "    .lazyPrintStatistics(\"< Main Output >\")\\\n",
    "    .link(\n",
    "        AkSinkBatchOp()\\\n",
    "            .setFilePath(DATA_DIR + TRAIN_FILE)\\\n",
    "            .setOverwriteSink(True)\n",
    "    );\n",
    "\n",
    "spliter.getSideOutput(0)\\\n",
    "    .lazyPrintStatistics(\"< Side Output >\")\\\n",
    "    .link(\n",
    "        AkSinkBatchOp()\\\n",
    "            .setFilePath(DATA_DIR + TEST_FILE)\\\n",
    "            .setOverwriteSink(True)\n",
    "    );\n",
    "\n",
    "BatchOperator.execute();"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_3_1\n",
    "\n",
    "source = CsvSourceBatchOp()\\\n",
    "    .setFilePath(DATA_DIR + ORIGIN_FILE)\\\n",
    "    .setSchemaStr(SCHEMA_STRING);\n",
    "\n",
    "source.lazyPrintStatistics(\"< Origin data >\");\n",
    "\n",
    "scaler = StandardScaler().setSelectedCols(FEATURE_COL_NAMES);\n",
    "\n",
    "scaler\\\n",
    "    .fit(source)\\\n",
    "    .transform(source)\\\n",
    "    .lazyPrintStatistics(\"< after Standard Scale >\");\n",
    "\n",
    "BatchOperator.execute();\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_3_2\n",
    "\n",
    "source = CsvSourceBatchOp()\\\n",
    "    .setFilePath(DATA_DIR + ORIGIN_FILE)\\\n",
    "    .setSchemaStr(SCHEMA_STRING);\n",
    "\n",
    "source.lazyPrintStatistics(\"< Origin data >\");\n",
    "\n",
    "scaler = MinMaxScaler().setSelectedCols(FEATURE_COL_NAMES);\n",
    "\n",
    "scaler\\\n",
    "    .fit(source)\\\n",
    "    .transform(source)\\\n",
    "    .lazyPrintStatistics(\"< after MinMax Scale >\");\n",
    "\n",
    "BatchOperator.execute();\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_3_3\n",
    "\n",
    "source = CsvSourceBatchOp()\\\n",
    "    .setFilePath(DATA_DIR + ORIGIN_FILE)\\\n",
    "    .setSchemaStr(SCHEMA_STRING);\n",
    "\n",
    "source.lazyPrintStatistics(\"< Origin data >\");\n",
    "\n",
    "scaler = MaxAbsScaler().setSelectedCols(FEATURE_COL_NAMES);\n",
    "\n",
    "scaler\\\n",
    "    .fit(source)\\\n",
    "    .transform(source)\\\n",
    "    .lazyPrintStatistics(\"< after MaxAbs Scale >\");\n",
    "\n",
    "BatchOperator.execute();\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_4_1\n",
    "\n",
    "source = CsvSourceBatchOp()\\\n",
    "    .setFilePath(DATA_DIR + ORIGIN_FILE)\\\n",
    "    .setSchemaStr(SCHEMA_STRING)\\\n",
    "    .link(\n",
    "        VectorAssemblerBatchOp()\\\n",
    "            .setSelectedCols(FEATURE_COL_NAMES)\\\n",
    "            .setOutputCol(VECTOR_COL_NAME)\\\n",
    "            .setReservedCols([LABEL_COL_NAME])\n",
    "    );\n",
    "\n",
    "source.link(\n",
    "    VectorSummarizerBatchOp()\\\n",
    "        .setSelectedCol(VECTOR_COL_NAME)\\\n",
    "        .lazyPrintVectorSummary(\"< Origin data >\")\n",
    ");\n",
    "\n",
    "VectorStandardScaler()\\\n",
    "    .setSelectedCol(VECTOR_COL_NAME)\\\n",
    "    .fit(source)\\\n",
    "    .transform(source)\\\n",
    "    .link(\n",
    "        VectorSummarizerBatchOp()\\\n",
    "            .setSelectedCol(VECTOR_COL_NAME)\\\n",
    "            .lazyPrintVectorSummary(\"< after Vector Standard Scale >\")\n",
    "    );\n",
    "\n",
    "VectorMinMaxScaler()\\\n",
    "    .setSelectedCol(VECTOR_COL_NAME)\\\n",
    "    .fit(source)\\\n",
    "    .transform(source)\\\n",
    "    .link(\n",
    "        VectorSummarizerBatchOp()\\\n",
    "            .setSelectedCol(VECTOR_COL_NAME)\\\n",
    "            .lazyPrintVectorSummary(\"< after Vector MinMax Scale >\")\n",
    "    );\n",
    "\n",
    "VectorMaxAbsScaler()\\\n",
    "    .setSelectedCol(VECTOR_COL_NAME)\\\n",
    "    .fit(source)\\\n",
    "    .transform(source)\\\n",
    "    .link(\n",
    "        VectorSummarizerBatchOp()\\\n",
    "            .setSelectedCol(VECTOR_COL_NAME)\\\n",
    "            .lazyPrintVectorSummary(\"< after Vector MaxAbs Scale >\")\n",
    "    );\n",
    "\n",
    "BatchOperator.execute();"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_4_2\n",
    "\n",
    "source = CsvSourceBatchOp()\\\n",
    "    .setFilePath(DATA_DIR + ORIGIN_FILE)\\\n",
    "    .setSchemaStr(SCHEMA_STRING)\\\n",
    "    .link(\n",
    "        VectorAssemblerBatchOp()\\\n",
    "            .setSelectedCols(FEATURE_COL_NAMES)\\\n",
    "            .setOutputCol(VECTOR_COL_NAME)\\\n",
    "            .setReservedCols([LABEL_COL_NAME])\n",
    "    );\n",
    "\n",
    "source\\\n",
    "    .link(\n",
    "        VectorNormalizeBatchOp()\\\n",
    "            .setSelectedCol(VECTOR_COL_NAME)\\\n",
    "            .setP(1.0)\n",
    "    )\\\n",
    "    .firstN(5)\\\n",
    "    .print();\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_5\n",
    "\n",
    "df = pd.DataFrame(\n",
    "    [\n",
    "        [\"a\", 10.0, 100],\n",
    "        [\"b\", -2.5, 9],\n",
    "        [\"c\", 100.2, 1],\n",
    "        [\"d\", -99.9, 100],\n",
    "        [None, None, None]\n",
    "    ]\n",
    ")  \n",
    "source = BatchOperator\\\n",
    "    .fromDataframe(df, schemaStr='col1 string, col2 double, col3 double')\\\n",
    "    .select(\"col1, col2, CAST(col3 AS INTEGER) AS col3\")\n",
    "\n",
    "\n",
    "source.lazyPrint(-1, \"< origin data >\");\n",
    "\n",
    "pipeline = Pipeline()\\\n",
    "    .add(\n",
    "        Imputer()\\\n",
    "            .setSelectedCols([\"col1\"])\\\n",
    "            .setStrategy('VALUE')\\\n",
    "            .setFillValue(\"e\")\n",
    "    )\\\n",
    "    .add(\n",
    "        Imputer()\\\n",
    "            .setSelectedCols([\"col2\", \"col3\"])\\\n",
    "            .setStrategy('MEAN')\n",
    "    );\n",
    "\n",
    "pipeline.fit(source)\\\n",
    "    .transform(source)\\\n",
    "    .print();\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_6\n",
    "\n",
    "dict_arr = {'name':['Alice','Bob','Cindy'], \n",
    "            'value':[1,2,3]}\n",
    "\n",
    "pd.DataFrame(dict_arr)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "arr_2D =[\n",
    "    ['Alice',1],\n",
    "    ['Bob',2],\n",
    "    ['Cindy',3]\n",
    "]\n",
    "\n",
    "pd.DataFrame(arr_2D, columns=['name', 'value'] )  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "arr_2D =[\n",
    "    ['Alice',1],\n",
    "    ['Bob',2],\n",
    "    ['Cindy',3]\n",
    "]\n",
    "\n",
    "pd.DataFrame(arr_2D, columns=['name', 'value'] )  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# DataFrame <> BatchOperator\n",
    "source = CsvSourceBatchOp()\\\n",
    "    .setFilePath(\"http://archive.ics.uci.edu/ml/machine-learning-databases/iris/iris.data\")\\\n",
    "    .setSchemaStr(\"sepal_length double, sepal_width double, petal_length double, petal_width double, category string\")\n",
    "source.firstN(5).print()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_iris = source.collectToDataframe()\n",
    "df_iris.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "iris = BatchOperator\\\n",
    "    .fromDataframe(\n",
    "        df_iris,\n",
    "        \"sepal_length double, sepal_width double, petal_length double, \" \n",
    "        + \"petal_width double, category string\"\n",
    "    );\n",
    "\n",
    "iris.firstN(5).print()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}
